/**
 *    Copyright (c) [2022] [中和农信]
 *    rabbit-mq-plus is licensed under Mulan PSL v2.
 *    You can use this software according to the terms and conditions of the Mulan PSL v2.
 *    You may obtain a copy of Mulan PSL v2 at:
 *             http://license.coscl.org.cn/MulanPSL2
 *    THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND, EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT, MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
 *    See the Mulan PSL v2 for more details.
 *
 */
package com.cfpamf.athean.rabbit.mq.plus.common.utils;

import java.util.HashMap;
import java.util.Map;

import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.CustomExchange;
import org.springframework.amqp.core.Exchange;
import org.springframework.amqp.core.ExchangeBuilder;
import org.springframework.amqp.core.FanoutExchange;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageDeliveryMode;
import org.springframework.amqp.core.MessagePostProcessor;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.core.QueueBuilder;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;

import com.cfpamf.athean.rabbit.mq.plus.common.constant.BusinessConstants;
import com.cfpamf.athean.rabbit.mq.plus.common.enums.MessageTypeEnum;
import com.cfpamf.athean.rabbit.mq.plus.producer.core.RabbitMqPlusProducerResultTemplate;
import com.cfpamf.athena.utils.DateUtils;
import com.cfpamf.athena.utils.Tools;

import lombok.extern.slf4j.Slf4j;

/**
 * 
 * 类RabbitMqUtils.java的实现描述：
 *
 * @author Andy（陈洪桥）2021年09月23 19:39
 */
@Slf4j
public class RabbitMqUtils {

    /**
     * 消息id
     * 
     * @return
     */
    public static String messageId() {
        return Tools.uuid();
    }

    /**
     * 延迟交换机名称
     *
     * <pre>
     * 延迟交换机名称=前缀+应用名+消息类型+业务交换机名称（用应用名称区分隔离，补偿通知不干扰别的应用）
     * </pre>
     *
     * @param appName
     * @param exchangeName
     * @return
     */
    public static String buildDelayExchangeName(final MessageTypeEnum messageTypeEnum, final String appName,
        final String exchangeName) {
        return BusinessConstants.PREFIX_DELAY_RETRY.concat(appName).concat("-").concat(messageTypeEnum.toString())
            .concat("-").concat(exchangeName);
    }

    public static String buildDelayQueueName(final MessageTypeEnum messageTypeEnum, final String appName,
        final String queueName) {
        return BusinessConstants.PREFIX_DELAY_RETRY.concat(appName).concat("-").concat(messageTypeEnum.toString())
            .concat("-").concat(queueName);
    }

    public static Queue createQueue(final String queueName) {
        return QueueBuilder.durable(queueName).build();// 队列持久
    }

    public static Binding queueBindexchange(Queue queue, String exchangeName) {
        return BindingBuilder.bind(queue).to((FanoutExchange)ExchangeBuilder.fanoutExchange(exchangeName).build());
    }

    public static Binding queueBindexchange(final String queueName, String exchangeName) {
        Queue queue = createQueue(queueName);
        return BindingBuilder.bind(queue).to((FanoutExchange)ExchangeBuilder.fanoutExchange(exchangeName).build());
    }

    public static FanoutExchange createExchange(final String exchange) {
        return ExchangeBuilder.fanoutExchange(exchange).build();
    }

    public static Binding queueBindDelayExchange(final String queueName, final String exchangeName) {
        Queue queue = createQueue(queueName);
        Exchange exchange = createExchange(exchangeName);
        return queueBindDelayExchange(queue, exchange);
    }

    public static Binding queueBindDelayExchange(Queue queue, Exchange exchange) {
        return new BindingBuilder.GenericArgumentsConfigurer(BindingBuilder.bind(queue).to(exchange), "").noargs();
    }

    /**
     * 创建延迟交换机
     * 
     * @param exchange
     * @return
     */
    public static Exchange createDelayExchange(final String exchange) {
        Map<String, Object> args = new HashMap<>(1);
        // 设置交换机支持延迟消息推送消息类型
        args.put("x-delayed-type", "fanout");
        return new CustomExchange(exchange, "x-delayed-message", true, false, args);
    }

    /**
     * 发送消息
     * 
     * @param rabbitTemplate
     * @param exchange
     * @param messageId
     * @param data
     * @param rabbitMqPlusProducerResultTemplate
     * @param eventNo
     * @param eventTypeCode
     * @param retryCount
     */
    public static void sendMessage(final RabbitTemplate rabbitTemplate, final String exchange, final String messageId,
        final String data, final RabbitMqPlusProducerResultTemplate rabbitMqPlusProducerResultTemplate,
        final String eventNo, String eventTypeCode, final int retryCount) {
        sendDelayMessage(rabbitTemplate, exchange, messageId, 0, data, rabbitMqPlusProducerResultTemplate, eventNo,
            eventTypeCode, retryCount);
    }

    /**
     * 发送延迟消息
     * 
     * @param rabbitTemplate
     * @param exchange
     * @param messageId
     * @param delayMillisecond
     * @param data
     * @param rabbitMqPlusProducerResultTemplate
     * @param eventNo
     * @param eventTypeCode
     * @param retryCount
     */
    public static void sendDelayMessage(final RabbitTemplate rabbitTemplate, final String exchange,
        final String messageId, final int delayMillisecond, final String data,
        final RabbitMqPlusProducerResultTemplate rabbitMqPlusProducerResultTemplate, final String eventNo,
        String eventTypeCode, final int retryCount) {
        if (log.isDebugEnabled()) {
            log.debug("[rabbitMqPlus]延迟消息，messageId:{} data:{}", messageId, data);
        }
        CorrelationData correlationData = new CorrelationData(messageId);
        correlationData.getFuture().addCallback(result -> {
            log.info("[rabbitMqPlus]send message result {} {} {} {} isAck:{}", correlationData.getId(), eventNo,
                eventTypeCode, retryCount, result.isAck());
            // check结果
            if (result.isAck()) {
                // ACK
                rabbitMqPlusProducerResultTemplate.akc(messageId, eventNo, eventTypeCode);
                return;
            }
            // NACK
            rabbitMqPlusProducerResultTemplate.nack(messageId, eventNo, eventTypeCode, retryCount);
        }, ex -> {
            // FAIL
            log.info("[rabbitMqPlus]send message fail {} {} {} {} ", correlationData.getId(), eventNo, eventTypeCode,
                retryCount, ex.getMessage());
            rabbitMqPlusProducerResultTemplate.fail(messageId, eventNo, eventTypeCode, retryCount);
        });
        // 注意，在设置returnCallback的时候，必须要设置交换机处理失败消息的模式，设置为 true
        rabbitTemplate.setMandatory(true);

        // 发送消息时指定Header延迟时间
        rabbitTemplate.convertAndSend(exchange, null, data, postProcessMessage(messageId, delayMillisecond),
            correlationData);
        if (log.isDebugEnabled()) {
            log.debug("[rabbitMqPlus]延迟消息，messageId:{} {} {} data:{} 发送完成", messageId, eventNo, eventTypeCode, data);
        }
    }

    /**
     * 设置消息头
     * 
     * @param messageId
     * @param delayMillisecond
     * @return
     */
    private static MessagePostProcessor postProcessMessage(final String messageId, final int delayMillisecond) {
        return new MessagePostProcessor() {

            @Override
            public Message postProcessMessage(Message message) throws AmqpException {
                // 设置消息持久化
                message.getMessageProperties().setDeliveryMode(MessageDeliveryMode.PERSISTENT);
                if (delayMillisecond > BusinessConstants.ZERO) {
                    message.getMessageProperties().setDelay(delayMillisecond);
                }
                message.getMessageProperties().setMessageId(messageId);
                message.getMessageProperties().setTimestamp(DateUtils.nowTime());
                return message;
            }
        };
    }

}
